package chapter14

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * @author: 余辉  
 * @blog: https://blog.csdn.net/silentwolfyh
 * @date: 2024 - 08 - 31 9:28 下午
 * @description:
 * 需求：
 * 1、每个人评价最高的10部电影
 * 2、最热门的前50部电影 (被评分的次数说明热门程度）
 * 3、每个人的评分总和
 * 4、每部电影的总得分和平均得分
 * */
object MovieDSL {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("")
      .master("local[*]")
      .getOrCreate()

    import org.apache.spark.sql.functions._
    import spark.implicits._
    val df: DataFrame = spark.read.json("BookData/exercise/电影评分/rating.json").drop("raete")
    //    df.printSchema()
    //    df.show(10)

    /** *
     * 每个人评价最高的10部电影
     * 1、首先按照用户进行分区 ，按照评分进行排序（降序），且打上便签 ，（窗口函数）
     * 2、只需要获取 rn <11
     *
     * uid    rate  rn
     * uid1   5      1
     * uid1   4      2
     * uid1   4      3
     * uid1   3      4
     * uid1   3      5
     */
    val win = Window.partitionBy('uid).orderBy('rate.desc)
    df.select('uid, 'movie, 'rate, row_number() over (win) as "rn")
      .where("rn < 11")
      .show()

    /** *
     * 最热门的前50部电影 (被评分的次数说明热门程度）
     * 1、按照电影进行分组，求评分次数
     * 2、将评分次数进行排序（降序）
     */
    df.groupBy('movie).agg(count('movie) as "cm").orderBy('cm.desc).show()

    /** *
     * 每个人的评分总和
     */
    df.groupBy('uid).agg(sum('rate)).show()

    /** *
     * 每部电影的总得分和平均得分
     * 1、按照电影进行分组
     * 2、再在组内进行求sum , avg
     */
    df.groupBy('movie).agg(sum('rate), avg('rate)).show()
  }
}
